package org.example.rpc;

import com.google.common.base.Strings;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.example.util.PropertiesUtil;
import org.jboss.netty.util.internal.StringUtil;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;

/**
 * @ClassName RpcFactory
 * @Description: TODO
 * @Author mxzhang.ethan
 * @Date 2022/4/12
 * @Version V1.0
 **/
@Slf4j
public class RpcFactory {
    private static RpcFactory rpc;
    public static final String ZOOKEEPER_PATH_PREFIX = "/flag-rpc/";
    public static final String ZOOKEEPER_PATH_DELIMETER = "/";
    public static int SERVER_PORT = 9999;
    public static int ClIENT_PORT = 8888;
    public static String ZK_ADDR;
    private CuratorFramework zkClient;

    public static RpcFactory build() {
        if (rpc == null) {
            synchronized (RpcFactory.class) {
                if (rpc == null) {
                    rpc = new RpcFactory();
                }
            }
        }
        return rpc;
    }

    public static CuratorFramework zkClient() {
        return build().zkClient;
    }

    private RpcFactory() {
        //作为注册中心，zookeeper必有且地址固定
        ZK_ADDR = PropertiesUtil.read("rpc.properties", "rpc.zookeeper.address");
        zkClient = CuratorFrameworkFactory.newClient(ZK_ADDR, new ExponentialBackoffRetry(1000, 3));
        zkClient.start();
        String propertyStr = PropertiesUtil.read("rpc.properties", "rpc.server.port");
        if (!Strings.isNullOrEmpty(propertyStr)) {
            SERVER_PORT = Integer.parseInt(propertyStr);
        }
        propertyStr = PropertiesUtil.read("rpc.properties", "rpc.client.port");
        if (!Strings.isNullOrEmpty(propertyStr)) {
            ClIENT_PORT = Integer.parseInt(propertyStr);
        }
    }

    /**
     * 存放服务对象
     */
    private Map<String, Object> serviceMap = new HashMap<>();

    /**
     * server做两件事
     * 1，实例化对象，往注册中心Zookeeper挂个名字
     * 2，负责SocketServer用于给调用了接口的client传输对象
     * 注册服务，将serviceObj实现的所有接口都在注册中心注册了
     * @param serviceObj
     * @return
     */
    @SneakyThrows
    public RpcFactory register(Object serviceObj) {
        //注册所有接口名指向本服务serviceObj
        for (Class<?> anInterface : serviceObj.getClass().getInterfaces()) {
            String serviceName = anInterface.getTypeName();
            if (serviceMap.containsKey(serviceName)) {
                return this;
            }
            serviceMap.put(serviceName, serviceObj);
            String serviceZkPath = ZOOKEEPER_PATH_PREFIX
                    + serviceName
                    + ZOOKEEPER_PATH_DELIMETER
                    + InetAddress.getLocalHost().getHostAddress()
                    + ":"
                    + SERVER_PORT;
            //往zookeeper里注册个名字
            if (zkClient.checkExists().forPath(serviceZkPath) != null) {
                log.info("node[{}] is exist", serviceZkPath);
                return this;
            }
            //还没注册过，注册这个名字
            zkClient.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(serviceZkPath);
            log.info("create node[{}]", serviceZkPath);
        }
        return this;
    }

    /**
     * server端启动服务，收到请求就调用
     */
    public void start() {
        try (ServerSocket socketServer = new ServerSocket(SERVER_PORT);) {
            while (true) {
                String clientAddr = null;
                String clientRequestServiceName = null;
                try (Socket client = socketServer.accept();
                     ObjectInputStream objectInputStream = new ObjectInputStream(client.getInputStream());
                     ObjectOutputStream objectOutputStream = new ObjectOutputStream(client.getOutputStream())) {
                    //receive request and response
                    log.info("a new session connected:[{}]", client.getRemoteSocketAddress().toString());
                    Object requestObj = objectInputStream.readObject();
                    if (!(requestObj instanceof RpcRequest)) {
                        throw new RuntimeException("is not a request obj");
                    }
                    RpcRequest rpcRequest = (RpcRequest) requestObj;
                    //根据request中的类名找到对象，然后对象根据方法名，参数类型调用方法，将方法的结果返回给客户端
                    Object res = searchAndInvoke(rpcRequest);
                    objectOutputStream.writeObject(res);
                } catch (Exception e) {
                    log.error("为[{}]服务[{}]是出错：{}", clientAddr, clientRequestServiceName, e.getCause() == null ? e.getMessage() : e.getCause().toString());

                }
            }
        } catch (Exception e) {
            log.error("socketServer:{}", e.getCause() == null ? e.getMessage() : e.getCause().toString());
        }
    }

    private Object searchAndInvoke(RpcRequest rpcRequest) {
        String serviceName = rpcRequest.getTypeName();
        Object serviceObj = serviceMap.get(serviceName);
        RpcResponse rpcResponse = new RpcResponse();
        if (serviceObj == null) {
            rpcResponse.setStatus(RpcResponse.FAIL);
            return rpcResponse;
        }
        String methodName = rpcRequest.getMethodName();
        Class<?>[] argClaszs = rpcRequest.getParameterType();
        Object[] args = rpcRequest.getArgs();
        Method method;
        try {
            method = serviceObj.getClass().getMethod(methodName, argClaszs);
        } catch (NoSuchMethodException e) {
            //根据rpcRequest没找到方法，说明参数有误，返回
            rpcResponse.setStatus(RpcResponse.FAIL);
            return rpcResponse;
        }
        //没报错，说明找到了，调用后把方法的结果放入rpcResponse中
        try {
            rpcResponse.setStatus(RpcResponse.FIND);
            rpcResponse.setResponseObj(method.invoke(serviceObj, args));
        } catch (IllegalAccessException | InvocationTargetException e) {
            e.printStackTrace();
        }
        return rpcResponse;
    }

    /**
     * client做两件事：
     * 1，从注册中心Zookeeper里找到服务地址
     * 2，从服务地址获取对象
     *
     * @param serviceClasz
     * @param <T>
     * @return
     */
    @SneakyThrows
    public <T> T lookup(Class<T> serviceClasz) {
        return ProxyService.proxy(serviceClasz);
    }
}

